import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class FlinkCDCTest {
    public static void main(String[] args) throws Exception {
        //1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //todo 2.开启检查点 flink CDC将读取binlog的位置信息以状态的方式保存在ck，
        // 如果想要断点续传，需要从checkpoint或savepoint启动程序
        //2.1 开启checkpoint，每隔5s钟做一次ck，并指定ck的一致性语义
        env.enableCheckpointing(3000L, CheckpointingMode.EXACTLY_ONCE);
        //2.2 设置超时时间为1min
        env.getCheckpointConfig().setCheckpointTimeout(60*1000L);
        //2.3设置两次重启的最小时间间隔为3s
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);
        //2.4设置任务关闭的时候保留最后一次ck数据
        env.getCheckpointConfig().enableExternalizedCheckpoints(
                CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
        );
        //2.5 指定从ck自动重启策略
        env.setRestartStrategy(RestartStrategies.failureRateRestart(
                3, Time.days(1L),Time.minutes(1L)
        ));
        //2.6 设置状态后端
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage(
                "hdfs://hadoop102:8020/flinkCDC/220828"
        );
        //2.7 设置访问HDFS的用户名
        System.setProperty("HADOOP_USER_NAME","atguigu");

        //使用flinkCDC构建mysql source
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()//注意！！！ 这里的泛型不能丢
                .hostname("hadoop102")
                .port(3306)
                .databaseList("gmall")
                .tableList("gmall.base_trademark")//必须要写上库名
                .username("root")
                .password("123456")
                .startupOptions(StartupOptions.initial())
                .deserializer(new JsonDebeziumDeserializationSchema())
                .build();

        DataStreamSource<String> streamSource = env.fromSource(
                mySqlSource,
                WatermarkStrategy.noWatermarks(),
                "mysql-source");
        streamSource.print();
        env.execute();


    }
}
